package com.lew.jlight.redis;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Map;
import java.util.Set;

import redis.clients.jedis.BinaryJedis;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Pipeline;
import redis.clients.jedis.Tuple;
import redis.clients.jedis.exceptions.JedisConnectionException;
import redis.clients.jedis.exceptions.JedisException;
import redis.clients.util.Pool;

import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkState;

/**
 * RedisTemplate 提供了一个template方法，负责对Jedis连接的获取与归还。 {@code JedisAction<T>} 和 JedisActionNoResult两种回调接口，适用于有无返回值两种情况。
 * PipelineAction 与 PipelineActionResult两种接口，适合于pipeline中批量传输命令的情况。 <p> 同时提供一些JedisOperation中定义的 最常用函数的封装,
 * 如get/set/zadd等。
 */
public class RedisTemplate {

    private static Logger logger = LoggerFactory.getLogger(RedisTemplate.class);

    private Pool<Jedis> jedisPool;

    public RedisTemplate(Pool<Jedis> jedisPool) {
        this.jedisPool = jedisPool;
    }

    /**
     * Callback interface for template.
     */
    public interface JedisAction<T> {
        T action(Jedis jedis);
    }

    /**
     * Callback interface for template without result.
     */
    public interface JedisActionNoResult {
        void action(Jedis jedis);
    }

    /**
     * Callback interface for template.
     */
    public interface PipelineAction {
        List<Object> action(Pipeline Pipeline);
    }

    /**
     * Callback interface for template without result.
     */
    public interface PipelineActionNoResult {
        void action(Pipeline Pipeline);
    }

    /**
     * Execute with a call back action with result.
     */
    public <T> T execute(JedisAction<T> jedisAction) throws JedisException {
        return doExecute(jedisAction, 1);
    }

    private <T> T doExecute(JedisAction<T> jedisAction, int tryTimes) throws JedisException {
        checkEndless(tryTimes);
        checkNotNull(jedisAction, "parameter jedisAction must not be null");
        Jedis jedis = jedisPool.getResource();
        checkState(jedis != null, "jedis from pool should not be null");
        try {
            return jedisAction.action(jedis);
        } catch (JedisConnectionException e) {
            logger.info("jedis connection is closed by redis service, drop it");
            try {
                jedis.getClient().close();
            } catch (Exception ex) {
                logger.warn("fail to close jedis client, drop it");
            }
            return doExecute(jedisAction, tryTimes + 1);
        } finally {
            try {
                jedis.close();
            } catch (Exception ex) {
                // this should not happen, but anyway catch it to avoid risk
                logger.error("fail to close jedis", ex);
            }
        }
    }

    private void checkEndless(int tryTimes) {
        if (tryTimes > 500) {
            throw new JedisException("can not find available jedis, endless loop");
        }
    }

    /**
     * Execute with a call back action without result.
     */
    public void execute(JedisActionNoResult jedisAction) throws JedisException {
        doExecute(jedisAction, 1);
    }

    private void doExecute(JedisActionNoResult jedisActionNoResult, int tryTimes) throws JedisException {
        checkEndless(tryTimes);
        checkNotNull(jedisActionNoResult, "parameter jedisAction must not be null");
        Jedis jedis = jedisPool.getResource();
        checkState(jedis != null, "jedis from pool should not be null");
        try {
            jedisActionNoResult.action(jedis);
        } catch (JedisConnectionException e) {
            logger.info("jedis connection is closed by redis service, drop it");
            try {
                jedis.getClient().close();
            } catch (Exception ex) {
                logger.warn("fail to close jedis client, drop it", ex);
            }
            doExecute(jedisActionNoResult, tryTimes + 1);
        } finally {
            try {
                jedis.close();
            } catch (Exception e) {
                logger.error("fail to close jedis", e);
            }
        }
    }

    /**
     * Execute with a call back action with result in pipeline.
     */
    public List<Object> execute(PipelineAction pipelineAction) throws JedisException {
        return doExecute(pipelineAction, 1);
    }

    private List<Object> doExecute(PipelineAction pipelineAction, int tryTimes) throws JedisException {
        checkEndless(tryTimes);
        checkNotNull(pipelineAction, "parameter pipelineAction must not be null");
        Jedis jedis = jedisPool.getResource();
        checkState(jedis != null, "jedis from pool should not be null");
        try {
            Pipeline pipeline = jedis.pipelined();
            pipelineAction.action(pipeline);
            return pipeline.syncAndReturnAll();
        } catch (JedisConnectionException e) {
            logger.info("jedis connection is closed by redis service, drop it");
            try {
                jedis.getClient().close();
            } catch (Exception ex) {
                logger.warn("fail to close jedis client, drop it", ex);
            }
            return doExecute(pipelineAction, tryTimes + 1);
        } finally {
            try {
                jedis.close();
            } catch (Exception e) {
                logger.error("fail to close jedis", e);
            }
        }
    }

    /**
     * Execute with a call back action without result in pipeline.
     */
    public void execute(PipelineActionNoResult pipelineAction) throws JedisException {
        doExecute(pipelineAction, 1);
    }

    private void doExecute(PipelineActionNoResult pipelineAction, int tryTimes) throws JedisException {
        checkEndless(tryTimes);
        checkNotNull(pipelineAction, "parameter pipelineAction must not be null");
        Jedis jedis = jedisPool.getResource();
        checkState(jedis != null, "jedis from pool should not be null");
        try {
            Pipeline pipeline = jedis.pipelined();
            pipelineAction.action(pipeline);
            pipeline.sync();
        } catch (JedisConnectionException e) {
            logger.info("jedis connection is closed by redis service, drop it");
            try {
                jedis.getClient().close();
            } catch (Exception ex) {
                logger.warn("fail to close jedis client, drop it", ex);
            }

            doExecute(pipelineAction, tryTimes + 1);
        } finally {
            try {
                jedis.close();
            } catch (Exception e) {
                logger.error("fail to close jedis", e);
            }
        }
    }

    /**
     * Return the internal JedisPool.
     */
    public Pool<Jedis> getJedisPool() {
        return jedisPool;
    }


    // / Common Actions ///

    /**
     * Remove the specified keys. If a given key does not exist no operation is performed for this key. <p> return false
     * if one of the key is not exist.
     */
    public Boolean del(final String... keys) {
        return execute((JedisAction<Boolean>) jedis -> jedis.del(keys) == keys.length);
    }

    public Set<String> keys(String pattern) {
        return execute((JedisAction<Set<String>>) jedis -> jedis.keys(pattern));
    }

    /**
     * flushDB
     */
    public void flushDB() {
        execute((JedisActionNoResult) BinaryJedis::flushDB);
    }

    // / String Actions ///

    /**
     * Get the value of the specified key. If the key does not exist null is returned. If the value stored at key is not
     * a string an error is returned because GET can only handle string values.
     */
    public String get(final String key) {
        return execute((JedisAction<String>) jedis -> jedis.get(key));
    }

    public boolean exists(final String key) {
        return execute((JedisAction<Boolean>) jedis -> jedis.exists(key));
    }

    /**
     * Get the value of the specified key as Long.If the key does not exist null is returned.
     */
    public Long getAsLong(final String key) {
        String result = get(key);
        return result != null ? Long.valueOf(result) : null;
    }

    /**
     * Get the value of the specified key as Integer.If the key does not exist null is returned.
     */
    public Integer getAsInt(final String key) {
        String result = get(key);
        return result != null ? Integer.valueOf(result) : null;
    }

    /**
     * Get the values of all the specified keys. If one or more keys dont exist or is not of type String, a 'nil' value
     * is returned instead of the value of the specified key, but the operation never fails.
     */
    public List<String> mget(final String... keys) {
        return execute((JedisAction<List<String>>) jedis -> jedis.mget(keys));
    }

    /**
     * Set the string value as value of the key. The string can't be longer than 1073741824 bytes (1 GB).
     */
    public void set(final String key, final String value) {
        execute((JedisActionNoResult) jedis -> jedis.set(key, value));
    }


    /**
     * The command is exactly equivalent to the following group of commands: {@link #set(String, String) SET} + {@link
     * #expire(String, int) EXPIRE}. The operation is atomic.
     */
    public void setex(final String key, final String value, final int seconds) {
        execute((JedisActionNoResult) jedis -> jedis.setex(key, seconds, value));
    }

    /**
     * SETNX works exactly like {@link #set(String, String) SET} with the only difference that if the key already exists
     * no operation is performed. SETNX actually means "SET if Not exists". <p> return true if the key was set.
     */
    public Boolean setnx(final String key, final String value) {
        return execute((JedisAction<Boolean>) jedis -> jedis.setnx(key, value) == 1);
    }

    /**
     * The command is exactly equivalent to the following group of commands: {@link #setex(String, String, int) SETEX} +
     * {@link #set(String, String) SETNX}. The operation is atomic.
     */
    public Boolean setnxex(final String key, final String value, final int seconds) {
        return execute((JedisAction<Boolean>) jedis -> {
            String result = jedis.set(key, value, "NX", "EX", seconds);
            return RedisUtils.isStatusOk(result);
        });
    }

    /**
     * GETSET is an atomic set this value and return the old value command. Set key to the string value and return the
     * old value stored at key. The string can't be longer than 1073741824 bytes (1 GB).
     */
    public String getSet(final String key, final String value) {
        return execute((JedisAction<String>) jedis -> jedis.getSet(key, value));
    }

    /**
     * Increment the number stored at key by one. If the key does not exist or contains a value of a wrong type, set the
     * key to the value of "0" before to perform the increment operation. <p> INCR commands are limited to 64 bit signed
     * integers. <p> Note: this is actually a string operation, that is, in Redis there are not "integer" types. Simply
     * the string stored at the key is parsed as a base 10 64 bit signed integer, incremented, and then converted back
     * as a string.
     *
     * @return Integer reply, this commands will reply with the new value of key after the increment.
     */
    public Long incr(final String key) {
        return execute((JedisAction<Long>) jedis -> jedis.incr(key));
    }

    public Long incrBy(final String key, final long increment) {
        return execute((JedisAction<Long>) jedis -> jedis.incrBy(key, increment));
    }

    public Double incrByFloat(final String key, final double increment) {
        return execute((JedisAction<Double>) jedis -> jedis.incrByFloat(key, increment));
    }

    /**
     * Decrement the number stored at key by one. If the key does not exist or contains a value of a wrong type, set the
     * key to the value of "0" before to perform the decrement operation.
     */
    public Long decr(final String key) {
        return execute((JedisAction<Long>) jedis -> jedis.decr(key));
    }

    public Long decrBy(final String key, final long decrement) {
        return execute((JedisAction<Long>) jedis -> jedis.decrBy(key, decrement));
    }

    // / Hash Actions ///

    /**
     * If key holds a hash, retrieve the value associated to the specified field. <p> If the field is not found or the
     * key does not exist, a special 'nil' value is returned.
     */
    public String hget(final String key, final String fieldName) {
        return execute((JedisAction<String>) jedis -> jedis.hget(key, fieldName));
    }

    public List<String> hmget(final String key, final String... fieldsNames) {
        return execute((JedisAction<List<String>>) jedis -> jedis.hmget(key, fieldsNames));
    }

    public Map<String, String> hgetAll(final String key) {
        return execute((JedisAction<Map<String, String>>) jedis -> jedis.hgetAll(key));
    }

    public void hset(final String key, final String fieldName, final String value) {
        execute((JedisActionNoResult) jedis -> jedis.hset(key, fieldName, value));
    }

    public void hmset(final String key, final Map<String, String> map) {
        execute((JedisActionNoResult) jedis -> jedis.hmset(key, map));
    }

    public Boolean hsetnx(final String key, final String fieldName, final String value) {
        return execute((JedisAction<Boolean>) jedis -> jedis.hsetnx(key, fieldName, value) == 1);
    }

    public Long hincrBy(final String key, final String fieldName, final long increment) {
        return execute((JedisAction<Long>) jedis -> jedis.hincrBy(key, fieldName, increment));
    }

    public Double hincrByFloat(final String key, final String fieldName, final double increment) {
        return execute((JedisAction<Double>) jedis -> jedis.hincrByFloat(key, fieldName, increment));
    }

    public Long hdel(final String key, final String... fieldsNames) {
        return execute((JedisAction<Long>) jedis -> jedis.hdel(key, fieldsNames));
    }

    public Boolean hexists(final String key, final String fieldName) {
        return execute((JedisAction<Boolean>) jedis -> jedis.hexists(key, fieldName));
    }

    public Set<String> hkeys(final String key) {
        return execute((JedisAction<Set<String>>) jedis -> jedis.hkeys(key));
    }

    public Long hlen(final String key) {
        return execute((JedisAction<Long>) jedis -> jedis.hlen(key));
    }

    // / List Actions ///

    public Long lpush(final String key, final String... values) {
        return execute((JedisAction<Long>) jedis -> jedis.lpush(key, values));
    }

    public String rpop(final String key) {
        return execute((JedisAction<String>) jedis -> jedis.rpop(key));
    }


    public String brpop(final int timeout, final String key) {
        return execute((JedisAction<String>) jedis -> {
            List<String> nameValuePair = jedis.brpop(timeout, key);
            if (nameValuePair != null) {
                return nameValuePair.get(1);
            } else {
                return null;
            }
        });
    }

    /**
     * Not support for sharding.
     */
    public String rpoplpush(final String sourceKey, final String destinationKey) {
        return execute((JedisAction<String>) jedis -> jedis.rpoplpush(sourceKey, destinationKey));
    }

    /**
     * Not support for sharding.
     */
    public String brpoplpush(final String source, final String destination, final int timeout) {
        return execute((JedisAction<String>) jedis -> jedis.brpoplpush(source, destination, timeout));
    }

    public Long llen(final String key) {
        return execute((JedisAction<Long>) jedis -> jedis.llen(key));
    }

    public String lindex(final String key, final long index) {
        return execute((JedisAction<String>) jedis -> jedis.lindex(key, index));
    }

    public List<String> lrange(final String key, final int start, final int end) {
        return execute((JedisAction<List<String>>) jedis -> jedis.lrange(key, start, end));
    }

    public void ltrim(final String key, final int start, final int end) {
        execute((JedisActionNoResult) jedis -> jedis.ltrim(key, start, end));
    }

    public void ltrimFromLeft(final String key, final int size) {
        execute((JedisActionNoResult) jedis -> jedis.ltrim(key, 0, size - 1));
    }

    public Boolean lremFirst(final String key, final String value) {
        return execute((JedisAction<Boolean>) jedis -> jedis.lrem(key, 1, value) == 1);
    }

    public Boolean lremAll(final String key, final String value) {
        return execute((JedisAction<Boolean>) jedis -> jedis.lrem(key, 0, value) > 0);
    }

    // / Set Actions ///
    public Boolean sadd(final String key, final String member) {
        return execute((JedisAction<Boolean>) jedis -> jedis.sadd(key, member) == 1);
    }

    public Set<String> smembers(final String key) {
        return execute((JedisAction<Set<String>>) jedis -> jedis.smembers(key));
    }


    /**
     * return true for add new element, false for only update the score.
     */
    public Boolean zadd(final String key, final double score, final String member) {
        return execute((JedisAction<Boolean>) jedis -> jedis.zadd(key, score, member) == 1);
    }

    public Double zscore(final String key, final String member) {
        return execute((JedisAction<Double>) jedis -> jedis.zscore(key, member));
    }

    public Long zrank(final String key, final String member) {
        return execute((JedisAction<Long>) jedis -> jedis.zrank(key, member));
    }

    public Long zrevrank(final String key, final String member) {
        return execute((JedisAction<Long>) jedis -> jedis.zrevrank(key, member));
    }

    public Long zcount(final String key, final double min, final double max) {
        return execute((JedisAction<Long>) jedis -> jedis.zcount(key, min, max));
    }

    public Set<String> zrange(final String key, final int start, final int end) {
        return execute((JedisAction<Set<String>>) jedis -> jedis.zrange(key, start, end));
    }

    public Set<Tuple> zrangeWithScores(final String key, final int start, final int end) {
        return execute((JedisAction<Set<Tuple>>) jedis -> jedis.zrangeWithScores(key, start, end));
    }

    public Set<String> zrevrange(final String key, final int start, final int end) {
        return execute((JedisAction<Set<String>>) jedis -> jedis.zrevrange(key, start, end));
    }

    public Set<Tuple> zrevrangeWithScores(final String key, final int start, final int end) {
        return execute((JedisAction<Set<Tuple>>) jedis -> jedis.zrevrangeWithScores(key, start, end));
    }

    public Set<String> zrangeByScore(final String key, final double min, final double max) {
        return execute((JedisAction<Set<String>>) jedis -> jedis.zrangeByScore(key, min, max));
    }

    public Set<Tuple> zrangeByScoreWithScores(final String key, final double min, final double max) {
        return execute((JedisAction<Set<Tuple>>) jedis -> jedis.zrangeByScoreWithScores(key, min, max));
    }

    public Set<String> zrevrangeByScore(final String key, final double max, final double min) {
        return execute((JedisAction<Set<String>>) jedis -> jedis.zrevrangeByScore(key, max, min));
    }

    public Set<Tuple> zrevrangeByScoreWithScores(final String key, final double max, final double min) {
        return execute((JedisAction<Set<Tuple>>) jedis -> jedis.zrevrangeByScoreWithScores(key, max, min));
    }

    public Boolean zrem(final String key, final String member) {
        return execute((JedisAction<Boolean>) jedis -> jedis.zrem(key, member) == 1);
    }

    public Long zremByScore(final String key, final double start, final double end) {
        return execute((JedisAction<Long>) jedis -> jedis.zremrangeByScore(key, start, end));
    }

    public Long zremByRank(final String key, final long start, final long end) {
        return execute((JedisAction<Long>) jedis -> jedis.zremrangeByRank(key, start, end));
    }

    public void expire(final String key, final int second) {
        execute((JedisActionNoResult) jedis -> jedis.expire(key, second));
    }

    public Long zcard(final String key) {
        return execute((JedisAction<Long>) jedis -> jedis.zcard(key));
    }

    public void set(final byte[] key, final byte[] value) {
        execute((JedisActionNoResult) jedis -> jedis.set(key, value));
    }

    public void setex(final byte[] key, final byte[] value, final int seconds) {
        execute((JedisActionNoResult) jedis -> jedis.setex(key, seconds, value));
    }

    public byte[] get(final byte[] key) {
        return execute((JedisAction<byte[]>) jedis ->
                jedis.get(key));
    }
}
